package com.zx.mes.rocketmq.hyl.transaction;

import com.zx.mes.rocketmq.hyl.constant.Const;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * transaction producer
 *
 * @author 华云龙
 * @date 2018-12-3
 */
@Log4j2
public class TransactionProducer {


    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, InterruptedException {

///        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        TransactionMQProducer producer = new TransactionMQProducer(Const.TRANSACTION_ROCKET_MQ_GROUP_NAME);

        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), r -> {
            Thread thread = new Thread(r);
            thread.setName(Const.TRANSACTION_ROCKET_MQ_GROUP_NAME + "-check-thread");
            return thread;
        });

        producer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);

        producer.setExecutorService(executorService);

        TransactionListener transactionListener = new TransactionListenerImpl();
        producer.setTransactionListener(transactionListener);

        producer.start();

        Message message = new Message(Const.TRANSACTION_ROCKET_MQ_TOPIC, "transaction A", "key-transaction", "神不在的星期天!".getBytes(RemotingHelper.DEFAULT_CHARSET));

        TransactionSendResult sendResult = producer.sendMessageInTransaction(message, "我所带的参数!");
        log.info("当前发送状态:" + sendResult);
        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();

        //// 写以以下形式,好像有问题,改成以上写法
///        Thread.sleep(Integer.MAX_VALUE);

    }
}
